package com.atguigu.edu.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.base.BaseApp;
import com.atguigu.edu.realtime.common.bean.PlayCommentVideoBean;
import com.atguigu.edu.realtime.common.constant.Constant;
import com.atguigu.edu.realtime.common.function.BeanToJsonStrMapFunction;
import com.atguigu.edu.realtime.common.function.DimAsyncFunction;
import com.atguigu.edu.realtime.common.util.DateFormatUtil;
import com.atguigu.edu.realtime.common.util.FlinkSinkUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.concurrent.TimeUnit;

public class DwsPlayCommentPlayVideoWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsPlayCommentPlayVideoWindow()
                .start(22000,4,
                "dws_play_video", Constant.TOPIC_DWD_TRAFFIC_APPVIDEO);
    }
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaStrDS) {
        //提取出需要的数据封装为jsonobj
        SingleOutputStreamOperator<JSONObject> plyVdoJsonDs = kafkaStrDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String s, ProcessFunction<String, JSONObject>.Context context, Collector<JSONObject> collector) throws Exception {
                if (StringUtils.isNotEmpty(s) && s.length() > 0) {
                    JSONObject jsonObject = JSON.parseObject(s);
                    JSONObject appVideoJsonObj = jsonObject.getJSONObject("appVideo");
                    JSONObject commonJsonObj = jsonObject.getJSONObject("common");
                    Long ts = jsonObject.getLong("ts");
                    Long playSec = appVideoJsonObj.getLong("play_sec");
                    String videoId = appVideoJsonObj.getString("video_id");
                    String uid = commonJsonObj.getString("uid");
                    JSONObject playVideoJsonObj = new JSONObject();
                    playVideoJsonObj.put("play_sec", playSec);
                    playVideoJsonObj.put("video_id", videoId);
                    playVideoJsonObj.put("uid", uid);
                    playVideoJsonObj.put("ts", ts);
                    collector.collect(playVideoJsonObj);
                }
            }
        });
        KeyedStream<JSONObject, String> uidKeyedDs = plyVdoJsonDs.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                return jsonObject.getString("uid");
            }
        });
        SingleOutputStreamOperator<PlayCommentVideoBean> beanDs = uidKeyedDs.process(new KeyedProcessFunction<String, JSONObject, PlayCommentVideoBean>() {
            ValueState<String> lastPlayVideoDateState;

            @Override
            public void open(Configuration parameters) throws Exception {
                ValueStateDescriptor<String> valueStateDescriptor =
                        new ValueStateDescriptor<String>("lastplayVideoDateState", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(Time.days(1)).build());
                lastPlayVideoDateState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, KeyedProcessFunction<String, JSONObject, PlayCommentVideoBean>.Context context, Collector<PlayCommentVideoBean> collector) throws Exception {

                String videoId = jsonObject.getString("video_id");
                Long playSec = jsonObject.getLong("play_sec");
                Long uvCt = 0L;
                String lastPlayVideoDate = lastPlayVideoDateState.value();
                Long ts = jsonObject.getLong("ts");
                String curPlayDate = DateFormatUtil.tsToDate(ts);
                if (StringUtils.isEmpty(lastPlayVideoDate) || !lastPlayVideoDate.equals(curPlayDate)) {
                    uvCt = 1L;
                    lastPlayVideoDateState.update(curPlayDate);
                }
                PlayCommentVideoBean playCommentVideoBean = new PlayCommentVideoBean("", "", ""
                        , videoId, "", uvCt, playSec, ts);
                collector.collect(playCommentVideoBean);
            }
        });
        //uidKeyedDs.print();
        //beanDs.print();
        //关联章节维度
        SingleOutputStreamOperator<PlayCommentVideoBean> withChapterIdDs = AsyncDataStream.unorderedWait(beanDs, new DimAsyncFunction<PlayCommentVideoBean>() {
            @Override
            public void addDims(PlayCommentVideoBean obj, JSONObject dimJsonObj) {
                String chapterId = dimJsonObj.getString("chapter_id");
                obj.setChapterId(chapterId);
            }

            @Override
            public String getTableName() {
                return "dim_video_info";
            }

            @Override
            public String getRowKey(PlayCommentVideoBean obj) {
                String videoId = obj.getVideoId();
                return videoId;
            }
        }, 60, TimeUnit.SECONDS);
        //withChapterIdDs.print();

        //水位线
        SingleOutputStreamOperator<PlayCommentVideoBean> watermarksDs = withChapterIdDs.assignTimestampsAndWatermarks(WatermarkStrategy.<PlayCommentVideoBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<PlayCommentVideoBean>() {
            @Override
            public long extractTimestamp(PlayCommentVideoBean playCommentVideoBean, long l) {
                return playCommentVideoBean.getTs();
            }
        }));
        //watermarksDs.print();
        //按照维度分组
        KeyedStream<PlayCommentVideoBean, Tuple2<String, String>> keyedDs = watermarksDs.keyBy(new KeySelector<PlayCommentVideoBean, Tuple2<String, String>>() {
            @Override
            public Tuple2<String, String> getKey(PlayCommentVideoBean playCommentVideoBean) throws Exception {
                String videoId = playCommentVideoBean.getVideoId();
                String chapterId = playCommentVideoBean.getChapterId();
                return Tuple2.of(videoId, chapterId);
            }
        });
        //keyedDs.print();
        //开窗
        WindowedStream<PlayCommentVideoBean, Tuple2<String, String>, TimeWindow> window =
                keyedDs.window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)));
        //聚合
        SingleOutputStreamOperator<PlayCommentVideoBean> reduceDs = window.reduce(new ReduceFunction<PlayCommentVideoBean>() {
            @Override
            public PlayCommentVideoBean reduce(PlayCommentVideoBean playCommentVideoBean, PlayCommentVideoBean t1) throws Exception {
                playCommentVideoBean.setTotalSec(playCommentVideoBean.getTotalSec() + t1.getTotalSec());
                playCommentVideoBean.setUvCt(playCommentVideoBean.getUvCt() + t1.getUvCt());
                return playCommentVideoBean;
            }
        }, new WindowFunction<PlayCommentVideoBean, PlayCommentVideoBean, Tuple2<String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple2<String, String> stringStringTuple2, TimeWindow timeWindow, Iterable<PlayCommentVideoBean> iterable, Collector<PlayCommentVideoBean> collector) throws Exception {
                long start = timeWindow.getStart();
                long end = timeWindow.getEnd();
                String stt = DateFormatUtil.tsToDateTime(start);
                String edt = DateFormatUtil.tsToDateTime(end);
                String cur = DateFormatUtil.tsToDate(start);
                PlayCommentVideoBean playCommentVideoBean = iterable.iterator().next();
                playCommentVideoBean.setStt(stt);
                playCommentVideoBean.setEdt(edt);
                playCommentVideoBean.setCurDate(cur);
                collector.collect(playCommentVideoBean);
            }
        });
        reduceDs.print();
        //写入doris
        reduceDs.map(new BeanToJsonStrMapFunction<PlayCommentVideoBean>()).sinkTo(FlinkSinkUtil.getDorisSink("dws_play_comment_playvideo_window"));
        System.out.println("What can i say");


    }
}